package com.dj.ms.event;

import com.alibaba.fastjson.JSON;
import com.dj.ms.event.support.AsyncEventBusExecutor;
import com.google.common.collect.Queues;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <br>
 * <p>事件处理handle</p>
 *
 * <br>
 *
 * @author ZJG
 * @version 1.0
 * @date 2021/7/26 下午6:04
 */
@Slf4j
@Component
public class EventNoTaskBusHandler {

    /**
     * 同步事件
     */
    private final static EventBus syncEventBus = new EventBus("syncEventBus");

    /**
     * 异步事件
     */
    private final static AsyncEventBus asyncEventBus = new AsyncEventBus(
            new AsyncEventBusExecutor(5, 20),
            (exception, context) -> {
                if (Objects.nonNull(true)) {
                }
            });

    /**
     * 异步事件队列
     */
    private static final ThreadLocal<Queue<Object>> queue = ThreadLocal.withInitial(() -> Queues.newArrayDeque());

    /**
     * 是否在事务中
     */
    public static final ThreadLocal<Boolean> inTransaction = ThreadLocal.withInitial(() -> false);

    /**
     * 定时线程池
     */
    private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(
            5,
            new BasicThreadFactory
                    .Builder()
                    .namingPattern("e-retry-pool-%d")
                    .daemon(true)
                    .build());

    /**
     * 提交一个没有事务的事件,先提交同步事件，再提交异步事件
     *
     * @param event 事件对象
     */
    public static void postNoTransaction(Object event) {
        if (Objects.isNull(event)) {
            return;
        }
        String eventClassName = Objects.isNull(event) ? "null" : event.getClass().getSimpleName();
        postSyncEvent(event, eventClassName);
        asyncEventBus.post(event);
        if(log.isInfoEnabled()) {
            log.info("异步事件已发布，{}:{}", eventClassName, JSON.toJSON(event));
        }

    }

    /**
     * 提交一个事件,先提交同步事件，再提交异步事件
     *
     * @param event 事件对象
     */
    public static void post(Object event) {
        String eventClassName = Objects.isNull(event) ? "null" : event.getClass().getSimpleName();
        Queue<Object> queueForThread = queue.get();
        queueForThread.offer(event);
        if(log.isDebugEnabled()) {
            log.debug((!inTransaction.get() ? "未" : "已") + "在事务中，预存异步事件，{}:{}", eventClassName, JSON.toJSON(event));
        }

        if(!inTransaction.get()) {
            SCHEDULED_EXECUTOR_SERVICE.schedule(() -> {
                if(queueForThread.contains(event)) {
                    asyncEventBus.post(event);
                    log.info("async event submit success. {}", event);
                    queueForThread.remove(event);
                }
            }, 3, TimeUnit.SECONDS);
        }
    }

    /**
     * 注册同步监听器
     */
    public static void registerSync(Object object) {
        syncEventBus.register(object);
    }
    /**
     * 注册异步监听器
     */
    public static void registerAsync(Object object) {
        asyncEventBus.register(object);
    }

    /**
     * 发布同步事件
     * @param event
     * @param eventClassName
     */
    private static void postSyncEvent(Object event, String eventClassName) {
        try {
            syncEventBus.post(event);
        } catch (Error e) {
            log.error("同步事件异常，{}:{}", eventClassName, JSON.toJSON(event), e);
            throw e;
        }
    }


    /**
     * 触发队列中的异步事件
     */
    public static void doAsyncPost(){
        Queue<Object> queueForThread = queue.get();
        if(queueForThread == null || queueForThread.isEmpty()){
            return;
        }
        Object nextEvent = null;
        try {
            while ((nextEvent = queueForThread.poll()) != null) {
                asyncEventBus.post(nextEvent);
                if(log.isInfoEnabled()) {
                    log.info("事务完成，已触发异步事件，{}:{}", nextEvent.getClass().getSimpleName(), JSON.toJSON(nextEvent));
                }
            }
        }catch (Exception e){
            log.error("使用事务，异步事件触发异常，{}:{}", nextEvent != null ? nextEvent.getClass().getSimpleName() : "null", JSON.toJSON(nextEvent), e);
        }finally {
            queue.remove();
        }
    }

}
